package edu.purdue.cs.aggr.messageprocessor;

import edu.purdue.cs.aggr.AggrException;
import edu.purdue.cs.aggr.Main;
import edu.purdue.cs.aggr.VertexManager;
import edu.purdue.cs.aggr.message.AccumulatorMessage;
import edu.purdue.cs.aggr.message.Message;

public class AccumulatorMessageProcessor extends MessageProcessor {

	private VertexManager manager = null;
	static long firstAccumulatorReceivedTime = -1;
	static long receivedAccumulators = 0;
	
	public AccumulatorMessageProcessor (VertexManager node) {
		this.manager = node;
	}
	
	@Override
	public synchronized void processMessage(Message message) throws AggrException{
		System.out.println("Processing accumulator message from:" + message.getSender() + ", destination: " + message.getDestination());
		if (firstAccumulatorReceivedTime < 0) {
			firstAccumulatorReceivedTime = System.currentTimeMillis();
		}
		receivedAccumulators++;
		System.out.println("Accumulators. Expected: " + Main.fanout + "  Received:" + receivedAccumulators);
		if (receivedAccumulators == Main.fanout) {
			System.out.println("Time between receiving first and last accumulators :" + (System.currentTimeMillis() - firstAccumulatorReceivedTime) + "ms");
		}
		if (!(message instanceof AccumulatorMessage)) {
			throw new AggrException("Wrong type of message");
		}
		AccumulatorMessage accMessage = (AccumulatorMessage) message;
    System.out.println("Original source of message:" + accMessage.getAccumulator().getSourceId());
		manager.getVertex(message.getDestination()).addAccumulator(accMessage.getAccumulator());
	}
}
